package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.ExamCorrectPercentBean;
import com.atguigu.edu.realtime.common.bean.UserCommentBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import com.google.gson.JsonObject;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.util.Iterator;
import java.util.List;

public class DwsInteractionCommentWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsInteractionCommentWindow().start(13023,
                4,"DwsInteractionCommentWindow",
                Constant.TOPIC_DWD_INTERACTION_REVIEW_INFO);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s);
                    }
                }
        );
        SingleOutputStreamOperator<JSONObject> wmStream = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(
                        (v, ts) -> v.getLong("ts") * 1000
                ));



        SingleOutputStreamOperator<UserCommentBean> beanStream = wmStream.keyBy(v -> v.getString("course_id")).process(
                new ProcessFunction<JSONObject, UserCommentBean>() {

                    ListState<String> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<String>("ListStateDescriptor", String.class);
                        listState = getRuntimeContext().getListState(listStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, UserCommentBean>.Context context, Collector<UserCommentBean> collector) throws Exception {
                        Iterator<String> iterator = listState.get().iterator();
                        List<String> idList = IteratorUtils.toList(iterator);
                        Long userNum = 0L;
                        Long highUserNum = 0L;
                        Long totalScore = 0L;
                        totalScore += jsonObject.getLong("review_stars");
                        if (!idList.contains(jsonObject.getString("user_id"))) {
                            //独立用户
                            userNum++;
                            listState.add(jsonObject.getString("user_id"));
                            if (jsonObject.getLong("review_stars") > 3) {
                                highUserNum++;
                            }
                        }
                        UserCommentBean userCommentBean = UserCommentBean.builder()
                                .highUserNum(highUserNum)
                                .userNum(userNum)
                                .totalScore(totalScore)
                                .courseId(jsonObject.getString("course_id"))
                                .build();
                        collector.collect(userCommentBean);
                    }
                }
        );


        SingleOutputStreamOperator<UserCommentBean> reduceStream = beanStream.keyBy(v -> v.getCourseId()).window(
                TumblingProcessingTimeWindows.of(Time.seconds(10))
        ).reduce(
                new ReduceFunction<UserCommentBean>() {
                    @Override
                    public UserCommentBean reduce(UserCommentBean userCommentBean, UserCommentBean t1) throws Exception {
                        userCommentBean.setHighUserNum(userCommentBean.getHighUserNum() + t1.getHighUserNum());
                        userCommentBean.setTotalScore(userCommentBean.getTotalScore() + t1.getTotalScore());
                        userCommentBean.setUserNum(userCommentBean.getUserNum() + t1.getUserNum());
                        return userCommentBean;
                    }
                },
                new WindowFunction<UserCommentBean, UserCommentBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<UserCommentBean> iterable, Collector<UserCommentBean> collector) throws Exception {
                        UserCommentBean next = iterable.iterator().next();
                        next.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        next.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        next.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(next);
                    }
                }
        );

        SingleOutputStreamOperator<UserCommentBean> dimStream = reduceStream.map(
                new RichMapFunction<UserCommentBean, UserCommentBean>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHbaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public UserCommentBean map(UserCommentBean orderBean) throws Exception {
                        //根据流中的对象获取要关联的维度的主键
                        String courseId = orderBean.getCourseId();
                        //根据维度的主键获取对应的维度对象
                        JSONObject dimJsonObj = HBaseUtil.readRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_course_info", courseId, JSONObject.class);
                        //将维度对象相关的维度属性补充到流中的对象上
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setCourseName(dimJsonObj.getString("course_name"));

                        return orderBean;
                    }
                }
        );
        dimStream.print();


        dimStream.map(new BeanToJsonStrMapFunction())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_interaction_comment_window"));
    }
}
